package com.tte.common.utils.redis;

import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RLock;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * RedissonClientUtil
 */
@Component
@Slf4j
public class RedissonClientUtils {
    @Autowired
    private RedissonClient redissonClient;

    /**
     * 获取分布式锁
     *
     * @param lockKey     锁的键
     * @param expireTime   锁的过期时间
     * @param waitTimeout 等待获取锁的超时时间
     * @return 是否成功获取锁
     */
    public boolean lock(String lockKey, long expireTime ,long waitTimeout) {
        RLock lock = redissonClient.getLock(lockKey);
        try {
            return lock.tryLock(waitTimeout, expireTime, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            return false;
        }
    }

    public boolean lock(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        return lock.tryLock();

    }

    /**
     * 释放分布式锁
     *
     * @param lockKey 锁的键
     */
    public void unlock(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        if (lock.isHeldByCurrentThread()) {
            lock.unlock();
        }
    }


//    ----------------------------------------------------------------------------
//pom是使用redisson-spring-boot-starters     不需要配置类RedissonConfig.java
    /**
     * 发送消息到队列头部
     *
     * @param message
     */
    public void sendMessage(String key,String message) {
        try {
            RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque(key);
            if(null != blockingDeque && blockingDeque.size() > 1000){
                log.debug("队列数大于1000,等待5秒");
                Thread.sleep(5000);
            }
            blockingDeque.put(message);
        } catch (InterruptedException e) {
            log.error("异常：{}",e);
        }
    }

    /**
     * 从队列尾部阻塞读取消息，若没有消息，线程就会阻塞等待新消息插入，防止 CPU 空转
     */
    public String onMessage(String key) {
        RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque(key);
        try {
            return blockingDeque.take();
        } catch (InterruptedException e) {
            log.error("异常：{}",e);
            return null;
        }
    }

    /**
     * 从队列尾部阻塞读取消息，若没有消息，线程就会阻塞等待新消息插入，防止 CPU 空转
     */
    public String takeLast(String key) {
        RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque(key);
        try {
            return blockingDeque.takeLast();
        } catch (InterruptedException e) {
            log.error("异常：{}",e);
            return null;
        }
    }

    /**
     * 从队列尾部阻塞读取消息，若没有消息，线程就会阻塞等待新消息插入，防止 CPU 空转
     */
    public String takeFirst(String key) {
        RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque(key);
        try {
            return blockingDeque.takeFirst();
        } catch (InterruptedException e) {
//            log.error("异常：{}",e);
            return null;
        }
    }

    public long getSize(String key){
        RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque(key);
        if(null != blockingDeque && blockingDeque.size() > 2000){
            return blockingDeque.size();
        }
        return -1;
    }

    public List<String> pollFirst(String key, int size) throws InterruptedException {
        try {
            RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque(key);
            if (blockingDeque.isEmpty()) {
                Thread.sleep(60 * 1000);
                return null;
            }
            return blockingDeque.pollFirst(size);
        } catch (Exception e) {
//            log.error("异常：{}", e);
            return null;
        }
    }


}
